package com.itheima.kafka.simple;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;

/**
 * @Description: 消息生产者
 * @Version: V1.0
 */
public class ProducerFastStart {

    public static final String TOPIC = "itcast-heima";

    public static void main(String[] args) {

        // 创建 kafka 配置对象
        Properties properties = new Properties();

        // 参数封装**
//        properties.put("bootstrap.servers", "192.168.200.129:9092");
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.129:9092");
//        properties.put("acks", "all");
        // key 序列化对象StringSerializer
//        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // value 序列化 StringSerializer
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");


        // 创建 KafkaProducer 对象
        KafkaProducer producer = new KafkaProducer(properties);


        // 发消息
        try {
            for (int i = 0; i < 10; i++) {
                ProducerRecord record = new ProducerRecord<>(TOPIC, "00001","hello kafka !"+i);
//            producer.send(record);

                // 同步发送
//                RecordMetadata metadata = (RecordMetadata) producer.send(record).get();
//                System.out.println(metadata.offset());

                // 异步发送
                //发送消息
                try {
                    producer.send(record, new Callback() {
                        @Override
                        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                            if(e!=null){
                                e.printStackTrace();
                            }
                            System.out.println(recordMetadata.offset());
                        }
                    });
                }catch (Exception e){
                    e.printStackTrace();
                }
            }

        } catch (Exception ex) {
            ex.printStackTrace();
        }


        // 关闭消息
        producer.close();
    }

}
